package com.niit.spark.sparkStreaming

import com.alibaba.fastjson.JSON
import com.niit.utils.MyKafkaUtils
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.json4s.DefaultFormats
import org.json4s.jackson.Json

/**
 * @author 杨铭
 *         2022/11/19,14:54
 */

object sparkStreaming3 {
  case class Order(isCorrect: String, orderCategory: String, orderData: String, orderName: String, orderNum: Int)

  case class Result(orderNumber: String, correctNumber: Int, notCorrectNumber: Int)

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("sparkStreaming3")
    //每隔多少秒处理一次
    val sc = new StreamingContext(conf, Seconds(2))
    //取消日志
    sc.sparkContext.setLogLevel("ERROR")

    val streamRdd = MyKafkaUtils.getKafkaDStream(sc,"orders","niit03")
    val line = streamRdd.map(_.value())
//    val line = sc.socketTextStream(hostname = "localhost", port = 9999)
    line
      .filter(i => {
        //转化为Object
        try {
          JSON.parseObject(i, classOf[Order])
          true
        } catch {
          case ex: Exception => false
        }

      })
      // 转换数据格式
      .map(s => {
        val order = JSON.parseObject(s, classOf[Order])
        (order.orderCategory,order)
      })
      // 按照订单号分组
      .groupByKey()
      // 在分组里计算有效无效订单号，并且返回结果
      .map(t => {
        val oderNumber = t._1
        val list = t._2
        var correctNumber = 0
        var notCorrectNumber = 0
        list.foreach(order => {
          if ("Y".equals(order.isCorrect)) {
            correctNumber = correctNumber + 1
          } else {
            notCorrectNumber = notCorrectNumber + 1
          }
        })
        var map = Map(
          "oderNumber" -> oderNumber,
          "correctNumber" -> correctNumber,
          "notCorrectNumber" -> notCorrectNumber
        )
        val res = Json(DefaultFormats).write(map)
        res
        //        MyKafkaUtils.send("streaming3", res)
        //        Result(oderNumber, correctNumber, notCorrectNumber)
      }).foreachRDD(rdd => {
      rdd.collect().foreach(x => {
        MyKafkaUtils.send("streaming3", x)
//        println(x)
      })
    })
    //采集器开始
    sc.start()
    //等待采集器关闭
    sc.awaitTermination()

  }

}
